package com.tools.kafka;


import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.io.*;
import java.util.Properties;

/**
 * @author xin.ding
 * @date 2019/10/16 9:34
 */

public class SendFileTool {
    private static String filePath;
    private static String bootstrap;
    private static String topic;
    private static long delay;
    private static boolean isDisplay;

    public static void main(String[] args) {
        if (args.length == 0) {
            System.out.println("Usage: -file(file path) -bootstrap(kafka-Server) -topic(Topic) [-delay(time ms) -display]");
            return;
        }

        for (int i = 0; i < args.length; i++) {
            if ("-file".equalsIgnoreCase(args[i])) {
                filePath = args[i + 1];
            } else if ("-bootstrap".equalsIgnoreCase(args[i])) {
                bootstrap = args[i + 1];
            } else if ("-topic".equalsIgnoreCase(args[i])) {
                topic = args[i + 1];
            } else if ("-delay".equalsIgnoreCase(args[i])) {
                delay = Long.parseLong(args[i + 1]);
            } else if ("-display".equalsIgnoreCase(args[i])) {
                isDisplay = true;
            }
        }
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrap);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.LINGER_MS_CONFIG, "5");
        props.put(ProducerConfig.ACKS_CONFIG, "all");

        File file = new File(filePath);
        try (Producer<String, String> producer = new KafkaProducer<>(props);
             BufferedReader reader = new BufferedReader(new FileReader(file));) {
            String line = null;
            while ((line = reader.readLine()) != null) {
                ProducerRecord<String, String> record = new ProducerRecord<>(topic, line);
                producer.send(record);
                if (isDisplay) {
                    System.out.println("Send: " + line);
                }
                if (delay > 0) {
                    Thread.sleep(delay);
                }
            }

        } catch (FileNotFoundException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }


    }
}
